package com.afdor.rws.event.support;

import com.afdor.rws.annotation.extension.Extension;
import com.afdor.rws.event.AggregationEvent;
import com.afdor.rws.event.Event;
import com.afdor.rws.event.EventPersistenceHandler;
import com.afdor.rws.event.EventPersistenceManager;
import com.afdor.rws.event.config.EventConfig;
import com.afdor.rws.event.config.EventConfigUtils;
import lombok.EqualsAndHashCode;
import lombok.Getter;
import lombok.ToString;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.math.NumberUtils;

import java.util.ArrayList;
import java.util.BitSet;
import java.util.List;
import java.util.Objects;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 事件持久化管理器
 *
 * @author 悭梵
 * @date Created in 2018-08-20 14:25
 */
@Slf4j
@ToString
@EqualsAndHashCode
@Extension(value = "default,group", order = Extension.NORMAL_ORDER - 100, interfaceClass = EventPersistenceManager.class)
public class DefaultEventPersistenceManager implements EventPersistenceManager, Runnable {
    /**
     * 持久化事件最大线程数
     */
    private static final int EVENT_PERSISTENCE_THREAD_POOL_SIZE = 5;
    /**
     * 事件持久化管理器一次持久化事件数量
     */
    private int batchSize;
    @Getter
    private EventConfig config;

    @Getter
    private EventPersistenceHandler eventPersistenceHandler;

    @Getter
    private Queue<Event> eventQueue = new ConcurrentLinkedQueue<>();

    private BitSet eventExistsBitSet = new BitSet();

    private AtomicInteger continuousException = new AtomicInteger(0);

    /**
     * 事件持久化线程池
     */
    protected static final ScheduledThreadPoolExecutor EVENT_PERSISTENCE_THREAD_POOL = new ScheduledThreadPoolExecutor(EVENT_PERSISTENCE_THREAD_POOL_SIZE, new ThreadFactory() {
        /**
         * 线程组
         */
        private ThreadGroup threadGroup = new ThreadGroup("event-persistence") {
            {
                setDaemon(true);
                setMaxPriority(Thread.NORM_PRIORITY);
            }
        };

        private Thread.UncaughtExceptionHandler uncaughtExceptionHandler = new EventPersistenceUncaughtExceptionHandler();

        @Override
        public Thread newThread(Runnable runnable) {
            Thread thread = new Thread(threadGroup, runnable, "thread-event-persistence-" + EVENT_PERSISTENCE_THREAD_POOL.getQueue().size());
            thread.setDaemon(true);
            thread.setPriority(Thread.MIN_PRIORITY);
            thread.setUncaughtExceptionHandler(uncaughtExceptionHandler);
            return thread;
        }
    });

    @Override
    public DefaultEventPersistenceManager setConfig(EventConfig config) {
        this.config = config;
        return this;
    }

    @Override
    public DefaultEventPersistenceManager setEventPersistenceHandler(EventPersistenceHandler persistenceHandler) {
        this.eventPersistenceHandler = persistenceHandler;
        return this;
    }

    /**
     * 添加事件
     *
     * @param events
     */
    @Override
    public DefaultEventPersistenceManager addEvent(Event... events) {
        // 事件持久化处理器为空，不进行事件持久化
        if (getEventPersistenceHandler() == null) {
            return this;
        }
        for (int i = 0, length = events.length; i < length; i++) {
            Event event = events[i];
            // 不存在就添加
            if (!eventExistsBitSet.get(event.hashCode())) {
                // 添加事件
                eventQueue.add(event);
                // 设置状态位为已存在
                eventExistsBitSet.set(event.hashCode());
                // 添加子事件
                if (Objects.equals(Event.Type.AGGREGATION_ROOT, event.getType())) {
                    eventQueue.addAll(((AggregationEvent) event).getSource());
                }
            }
        }
        return this;
    }


    @Override
    public void run() {
        if (!Thread.currentThread().isInterrupted()) {
            /* 线程未终止，就执行事件消费。若持久化事件返回为true，这继续循环处理事件； 若处理事件返回为false，则退出执行，有计划任务重新启动执行*/
            while (consume()) {
                if (Thread.currentThread().isInterrupted()) {
                    break;
                }
            }
        }
    }

    @Override
    public DefaultEventPersistenceManager init() {
        batchSize = NumberUtils.createInteger(EventConfigUtils.getProperty(getConfig(), EventConfigUtils.EVENT_PERSISTENCE_BATCH_SIZE, "50"));
        int period = NumberUtils.createInteger(EventConfigUtils.getProperty(getConfig(), EventConfigUtils.EVENT_PERSISTENCE_PERIOD, "1000"));
        if (period <= 0) {
            period = 1000;
        }
        EVENT_PERSISTENCE_THREAD_POOL.scheduleWithFixedDelay(this, 0, period, TimeUnit.MILLISECONDS);
        return this;
    }

    @Override
    public void destroy() {
        EVENT_PERSISTENCE_THREAD_POOL.shutdown();
    }

    /**
     * 执行具体事件持久化操作
     *
     * @return
     */
    protected boolean consume() {
        // 允许事件持久化处理器为空，为空即表示事件的持久化不由系统复杂
        EventPersistenceHandler eventPersistenceHandler = getEventPersistenceHandler();
        if (eventPersistenceHandler != null) {

            /* 获取本次需要持久化的事件列表 */
            List<Event> batchEvent = new ArrayList<>(EVENT_PERSISTENCE_THREAD_POOL_SIZE);
            while (true) {
                Event event = eventQueue.poll();
                if (event != null) {
                    /* 已从队列中移除，清除状态 */
                    eventExistsBitSet.clear(event.hashCode());
                    batchEvent.add(event);
                    if (batchEvent.size() >= EVENT_PERSISTENCE_THREAD_POOL_SIZE) {
                        break;
                    }
                } else {
                    break;
                }
            }
            // 决定是否新一次性开线程持久化事件，不超过线程
            int poolSize = EVENT_PERSISTENCE_THREAD_POOL.getQueue().size();
            if (poolSize < EVENT_PERSISTENCE_THREAD_POOL_SIZE && eventQueue.size() > batchSize * 5 * poolSize) {
                EVENT_PERSISTENCE_THREAD_POOL.submit(this);
            }


            try {
                eventPersistenceHandler.persistence(batchEvent);
                continuousException.getAndSet(0);
            } catch (Exception e) {
                addEvent((Event[]) batchEvent.toArray());
                if (continuousException.incrementAndGet() > 10) {
                    // TODO 连续持久化异常警告，及处理策略
                }
                // 若这里持续报异常，可能会出现事件队列满溢出的BUG或频繁GC，导致OutOfMemoryError。
                log.error("持久化事件异常，已将事件重新放入队列，稍后继续执行持久化", e);
            }
        }
        return eventQueue.size() > 0;
    }

    /**
     * 线程异常拦截输出
     */
    public static class EventPersistenceUncaughtExceptionHandler implements Thread.UncaughtExceptionHandler {

        @Override
        public void uncaughtException(Thread t, Throwable e) {
            log.error("thread " + t.getName() + " exception", e);
        }
    }
}
